package hyl.core.db;

import java.sql.Array;
import java.sql.CallableStatement;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.ResultSetMetaData;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import hyl.core.MyFun;
import hyl.core.run.IDoMap;
import hyl.core.run.MyTimer;
import java.sql.Statement;

public class MyDB implements IDb {
	private static final Logger log = LoggerFactory.getLogger(MyDB.class);
	// 需求 1:单语句,自动提交和关闭 ,不可回滚,每次实例化重新连接
	public static final int OP_AUTO = 1;
	// 需求 2:散列事务,分布执行,手动提交,手动关闭 ,允许回滚 ,每次实例化重新连接
	public static final int OP_COMMIT = 2;
	// 需求 3:批量事务,批量提交,手动关闭,允许异常回滚,每次实例化重新连接
	// 需求3 可以用OP_AUTO 实现,也可以用OP_COMMIT 实现(后者可以连续调用)
	// 需求 4:后台定时事务处理,分布执行,定时提交,不允许回滚,手动关闭,每次实例化重新连接
	public static final int OP_TIMER = 3;
	// private PreparedStatement pst = null;
	private Statement _st = null;
	private ResultSet _rs = null;
	private MyConnection _myconn = null;
	// private boolean _autoCommit = true;
	// 不允许重新连接,因为statement resultset 状态都没有重置
	// 00 :不回滚+手动提交+自动关闭 1: 自动关闭,不回滚 2
	private String _dbalias; // 配置中的别名
	private int _type = OP_AUTO;
	private int _commitrun = 0;// 0:不在运行 1：正在运行
	public AtomicInteger _commitcount = new AtomicInteger(0);
	public boolean _transactionError = false;// 事务运行是否出错，出错的事务，后续脚本无需执行
	public IDoMap<MyDB> _异常通知 = null;

	/** alias :设置可选的别名，jdbc.json配置中的别名 */
	public static MyDB createDB(String alias) {
		return new MyDB(alias, OP_AUTO);
	}

	/**
	 * 默认是手动提交模式
	 */
	public static MyDB createCommitDB(String alias) {
		return new MyDB(alias, OP_COMMIT);
	}

	/** 每间隔一定时间period(单位:毫秒)自动提交 */
	public static MyDB createTimerDB(String alias, int PERIOD_MS) {
		MyDB db = new MyDB(alias, OP_TIMER);
		db.timeoutCommit(PERIOD_MS);
		return db;
	}

	/** 每间隔一定时间period(单位:毫秒)自动提交 默认每2秒提交 */
	public static MyDB createTimerDB(String alias) {
		MyDB db = new MyDB(alias, OP_TIMER);
		db.timeoutCommit(null);
		return db;
	}

	/** alias :设置可选的别名，jdbc.json配置中的别名 */
	private MyDB(String dbalias, Integer type) {
		_dbalias = dbalias;
		if (type != null && type > 0 && type < 4)
			_type = type;
	}

	/** 如果是自动提交模式，那么执行之前新建一个连接 */
	private void setConn() {
		if (_type == OP_AUTO) {
			_myconn = DBPool.getInstance().getConnection(_dbalias);
			try {
				if (!_myconn.getAutoCommit())
					_myconn.setAutoCommit(true);
			} catch (Exception e) {
				on连接异常("设置自动提交属性时连接异常", e);
			}
		} else {// 如果停止状态 重新启动
			if (_commitrun == 0) {
				_myconn = DBPool.getInstance().getConnection(_dbalias);
				try {
					if (_myconn.getAutoCommit())
						_myconn.setAutoCommit(false);
				} catch (Exception e) {
					on连接异常("设置手动提交属性时连接异常", e);
				}
				_commitrun = 1;
			}
		}
	}

	/**
	 * 更换别名 当数据库无法连接时,更换连接
	 * 
	 * @param alias
	 */
	public void setAlias(String alias) {
		_dbalias = alias;
		_myconn = DBPool.getInstance().getConnection(_dbalias);
	}

	// 获取当前连接
	public MyConnection getConn() {
		return _myconn;
	}

	public int getType() {
		return _type;
	}

	// 如果是自动提交，不允许回滚
	private void on连接异常(String msg, Exception e) {
		log.error(msg, e);
		if (_异常通知 != null)
			try {
				_异常通知.run(_dbalias, this);
			} catch (Exception e1) {
			}
		// _conn.fv重新连接();//不能加这一句
	}

	private void on异常(String msg, Exception e) {
		log.error(msg, e);
		// _conn.fv重新连接();//不能加这一句
	}

	private void on执行异常(String msg, Exception e) {
		log.error(msg, e);
		// e.printStackTrace();
		// 如果是自动提交 OP_AUTO 无需回滚 报异常但不处理
		// 如果是手动提交 记录异常 ,等待commit时候全部回滚
		if (_type > OP_AUTO) {
			isCommitConnErr(e);
			_transactionError = true;
		}
	}

	/**
	 * 断网时，auto 会阻塞线程直到连接打开， 而commit前的执行脚本会出现异常时， 同时会锁死线程，抛出conn=null
	 */
	private void isCommitConnErr(Exception e) {

		if (e.getMessage() == null || e.getMessage().contains("Connection reset")) {
			log.error("数据库服务器或网络无法连通导致sql脚本无法执行");
			MyDataSource ds = DBPool.getInstance().getDataSource(_dbalias);
			if (ds.testConn()) {
				String sql = null;
				String kill = null;
				boolean fl = true;
				int idr = 1;
				if (ds.getJdbcDriver().contains("mysql") || ds.getJdbcDriver().contains("mariadb")) {
					sql = "select trx_mysql_thread_id id FROM information_schema.INNODB_TRX";
					idr = 1;
				} else if (ds.getJdbcDriver().contains("sqlserver") || ds.getJdbcDriver().contains("jtds")) {
					sql = "select request_session_id id from sys.dm_tran_locks where resource_type='OBJECT'";
					idr = 2;
				} else if (ds.getJdbcDriver().contains("oracle")) {
					sql = "select concat(serial#,osuser) id  from v$session a,gv$lock b where a.sid=b.sid ";
					idr = 3;
				} else {
					fl = false;// 未知的驱动不处理
				}
				if (fl) {
					MyDB db = createDB(_dbalias);
					List<Map<String, Object>> s = db.queryList(sql, null);
					List<String> arr = new ArrayList<>();
					for (Map<String, Object> a : s) {
						kill = null;
						switch (idr) {
						case 1:
							java.math.BigInteger b = (java.math.BigInteger) a.get("id");
							kill = "kill " + b.longValue();
							break;
						case 2:
							int b2 = (int) a.get("id");
							kill = "kill " + b2;
							break;
						case 3:
							String b3 = (String) a.get("id");
							kill = "alter system kill session '" + b3 + "'";
							break;
						}
						if (kill != null)
							arr.add(kill);
					}
					if (arr.isEmpty())
						return;
					db.batch(arr);
				}
			}
		}
	}

	public boolean isAuto() {
		return _type == OP_AUTO;
	}

	/** 如果是事务并且之前执行出现异常，那么终止该事务的后续脚本执行 */
	private boolean isStop() {
		if (_type > OP_AUTO && _transactionError)
			return true;
		return false;
	}

	/**
	 * <p>
	 * Title:用于commitDb 方式 手动提交
	 * </p>
	 * <p>
	 * Description: 不适用于批命令batch和事务命令trancation 而且必须和createTranscation 配套使用 该方法主要用于
	 * query** 和execute** 开头的函数 当多个sql 一起执行结束,可以使用该方法一起提交到数据库
	 * 如果不提交,也可以查到数据,但是只从内存中取,不从数据库取数.
	 * </p>
	 * 
	 * @return :void @throws
	 */
	public synchronized boolean commit() {
		if (_type == OP_AUTO || _myconn == null) { // 非正常调用,自动过滤 不处理
			return true;
		}
		boolean flag = false;
		if (_myconn.isExit()) {
			log.error("事务提交前连接出现非正常关闭, 有关的任务脚本可能需要重新执行");
		} else if (_commitcount.intValue() > 0) {// 如果没有提交物不触发提交
			try {
				if (_transactionError) {// 出现执行错误,回滚
					_myconn.rollback();
				} else// 否则提交
				{
					_myconn.commit();
					flag = true;
				}
			} catch (Exception e) {
				flag = false;
				on异常("mydb.commit() [OP_COMMIT] ", e);
			}
		}
		closeAll();
		_commitrun = 0;// 不在运行，停止状态
		return flag;
	}

	// 关闭结果集对象;
	private synchronized void closeRs() {
		// 关闭结果集对象
		if (_rs != null) {
			try {
				_rs.close();
			} catch (Exception e) {
				on异常("rs.close()时;", e);
			}
		}
		// 关闭Statement对象
		_rs = null;
	}

	private synchronized void closeSt() {
		if (_st != null) {
			try {
				_st.close();
			} catch (Exception e) {
				on异常("st.close()时;", e);
			}
		}
		_st = null;
	}

	/**
	 * closeall 是指关闭所有用到的 result， statement conntion 对象
	 * 如果是定时任务型的实例化对象，不允许关闭，但是可以调用commit 手动提交 关闭所有资源
	 */
	private synchronized void closeAll() {
		// 防止关闭异常
		_transactionError = false;
		if (_myconn == null || _myconn.isExit()) {
			return;
		}
		_commitcount.set(0);
		// 关闭结果集对象
		closeRs();
		closeSt();
		// 关闭Connection 对象，返回到连接池
		// 取出用,用后关闭,关闭后,连接池没有追补
		_myconn.close();// 关闭即归还连接
		// _conn.showDataSourceInfo() ;
	}

	private void autoCloseAll() {
		if (_type == OP_AUTO) {
			closeAll();
		}
	}

	// 试用与in的参数
	/*
	 * in 有两种写法 一种 in ? 另外一种 in [?,?,?...] 第一种需要用到以下函数来获取Array对象 oracle 是试用的,但是mysql
	 * 好像不适用
	 */
	public Array getInArray(String intype, Object[] objes) throws SQLException {
		return _myconn.createArrayOf(intype, objes);
	}

	/**
	 * SQL 查询将查询结果直接放入ResultSet中
	 * 
	 * @param sql    SQL语句
	 * @param params 参数数组，若没有参数则 为null
	 * @return 结果集
	 * @throws SQLException
	 */
	private synchronized ResultSet queryRs(String sql, Object[] params) throws SQLException {
		setConn();
		closeRs();
		// 参数赋值
		if (params == null || params.length == 0) {
			_st = _myconn.createStatement();
			_rs = _st.executeQuery(sql);
		} else {
			PreparedStatement pst = _myconn.prepareStatement(sql);
			// System.out.println("queryRs() pst close="+pst.isClosed());
			_st = pst;
			for (int i = 0; i < params.length; i++) {
				if (params[i] instanceof Array) {
					pst.setArray(i + 1, (Array) params[i]);
					// System.out.println("suzu");
				} else
					pst.setObject(i + 1, params[i]);
			}
			// 执行
			_rs = pst.executeQuery();
		}
		// _commitcount.incrementAndGet();//commit 主要用于 执行事务 不适合查询select 不需要计数
		return _rs;
	}

	/**
	 * SQL 查询将查询结果： 只查询一个字段值
	 * 
	 * @param sql    SQL语句
	 * @param params 参数数组，若没有参数则为null
	 * @return 结果集
	 */
	public synchronized Object querySingle(String sql, Object[] params) {
		if (isStop())
			return null;
		Object object = null;
		try {
			_rs = queryRs(sql, params);
			// 执行
			if (_rs.next()) {
				object = _rs.getObject(1);
			}
		} catch (Exception e) {
			on执行异常("querySingle() =" + sql, e);
		} finally {
			autoCloseAll();
		}
		return object;
	}

	/**
	 * 获取（只有）一条结果集，并将结果放在map中
	 * 
	 * @param sql SQL语句 params 参数，没有则为null
	 * @return map 只取第一行结果
	 */
	public synchronized Map<String, Object> queryMap(String sql, Object[] params) {
		if (isStop())
			return null;
		// 执行SQL获得结果集
		// 创建ResultSetMetaData对象
		ResultSetMetaData rsmd = null;
		// 结果集列数
		int columnCount = 0;
		// 创建List
		Map<String, Object> map = new HashMap<String, Object>();
		try {
			_rs = queryRs(sql, params);
			rsmd = _rs.getMetaData();
			// 获得结果集列数
			columnCount = rsmd.getColumnCount();
			// 将ResultSet的结果保存到List中
			if (_rs.next()) {
				for (int i = 1; i <= columnCount; i++) {
					map.put(rsmd.getColumnLabel(i), _rs.getObject(i));
				}
			}
		} catch (Exception e) {
			on执行异常("queryMap()=" + sql, e);
		} finally {
			autoCloseAll();
		}
		return map;
	}

	/**
	 * 获取多条结果集，并将结果放在List中 每一条记录是map类型 也就是包含了字段名，所以如果数据多的话，字段名冗余的厉害
	 * 
	 * @param sql SQL语句 params 参数，没有则为null
	 * @return List 结果集
	 */
	public synchronized List<Map<String, Object>> queryList(String sql, Object[] params) {
		if (isStop())
			return null;
		// 执行SQL获得结果集
		// 创建ResultSetMetaData对象
		ResultSetMetaData rsmd = null;
		// 结果集列数
		int columnCount = 0;
		// 创建List
		List<Map<String, Object>> list = new ArrayList<Map<String, Object>>();
		try {
			_rs = queryRs(sql, params);
			rsmd = _rs.getMetaData();
			// 获得结果集列数
			columnCount = rsmd.getColumnCount();
			// 将ResultSet的结果保存到List中
			while (_rs.next()) {
				Map<String, Object> map = new HashMap<String, Object>();
				for (int i = 1; i <= columnCount; i++) {
					map.put(rsmd.getColumnLabel(i), _rs.getObject(i));
				}
				list.add(map);// 每一个map代表一条记录，把所有记录存在list中
			}
		} catch (Exception e) {
			on执行异常("queryList(); sql=" + sql, e);
		} finally {
			autoCloseAll();
		}
		return list;
	}

	/**
	 * 查询结果 外list 为记录集合 内list是 每一条记录， 由所有字段值组成的数组集合，不包含列名，为了减少返回数据，主要用于json
	 * 
	 * @param sql
	 * @param params
	 * @return
	 */
	public synchronized List<List<Object>> queryList2D(String sql, Object[] params) {
		if (isStop())
			return null;
		// 创建ResultSetMetaData对象
		ResultSetMetaData rsmd = null;
		// 结果集列数
		int columnCount = 0;
		// 创建List
		List<List<Object>> list = new ArrayList<List<Object>>();
		try { // 执行SQL获得结果集
			_rs = queryRs(sql, params);
			rsmd = _rs.getMetaData();
			// 获得结果集列数
			columnCount = rsmd.getColumnCount();
			// 将ResultSet的结果保存到List中
			while (_rs.next()) {
				List<Object> record = new ArrayList<Object>();
				for (int i = 1; i <= columnCount; i++) {
					record.add(_rs.getObject(i));
				}
				list.add(record);// 每一个map代表一条记录，把所有记录存在list中
			}
		} catch (Exception e) {
			on执行异常("queryList2D(); sql=" + sql, e);
		} finally {
			autoCloseAll();
		}
		return list;
	}

	/**
	 * 存储过程带有一个输出参数的方法
	 * 
	 * @param sql         存储过程语句
	 * @param params      参数数组
	 * @param outParamPos 输出参数位置
	 * @param SqlType     输出参数类型 从左往右 第一个参数位置是1
	 * @return 输出参数的值
	 */
	public synchronized Object excuteFun(String sql, Object[] inparams) {
		return excuteFun(sql, inparams, null, null);
	}

	public synchronized Object excuteFun(String sql, Object[] params, Integer outParamPos, Integer outType) {
		if (isStop())
			return null;
		Object object = null;
		CallableStatement cst;
		setConn();
		try {
			// 调用存储过程
			// prepareCall:创建一个 CallableStatement 对象来调用数据库存储过程。
			_st = cst = _myconn.prepareCall(sql);
			// 给参数赋值
			if (params != null) {
				for (int i = 0; i < params.length; i++) {
					cst.setObject(i + 1, params[i]);
				}
			}
			// 注册输出参数
			if (outParamPos != null && outType != null) {
				// callableStatement.registerOutParameter(2, Types.NUMERIC);
				cst.registerOutParameter(outParamPos, outType);
			}
			// 执行
			cst.execute();
			// 得到输出参数
			if (outParamPos != null)
				object = cst.getObject(outParamPos);
			_commitcount.incrementAndGet();
		} catch (Exception e) {
			on执行异常("excuteFun(); sql=" + sql, e);
		}
		autoCloseAll();
		return object;
	}

	/**
	 * 存储过程带有一个输出参数的方法 不适用于多个语句连续提交的场景
	 * 
	 * @param sql      sql语句 :insert update delete truncate create
	 * @param bindArgs 参数数组
	 * @return 输出参数的值 执行条数
	 */
	public synchronized Long executeUpdate(String sql, Object[] bindArgs) {
		if (isStop())
			return -1L;
		/** 影响的行数 **/
		Long ID = 0L;
		setConn();
		try {
			/** 从数据库连接池中获取数据库连接 **/
			/** 执行SQL预编译 **/
			/** 设置不自动提交，以便于在出现异常的时候数据库回滚 **/
			PreparedStatement pst;
			_st = pst = _myconn.prepareStatement(sql, Statement.RETURN_GENERATED_KEYS);
			if (bindArgs != null) {
				/** 绑定参数设置sql占位符中的值 **/
				for (int i = 0; i < bindArgs.length; i++) {
					pst.setObject(i + 1, bindArgs[i]);
				}
			}
			/** 执行sql **/
			pst.executeUpdate();
			ResultSet rs = pst.getGeneratedKeys(); // 获取结果
			if (rs.next()) {
				ID = rs.getLong(1);// 取得ID
			}
			_commitcount.incrementAndGet();
		} catch (Exception e) {
			on执行异常("executeUpdate(sql,[]);conn=" + _myconn.getId() + " ，conn=" + _myconn._i状态 + ",sql=" + sql, e);
			ID = -1L;
		}
		autoCloseAll();
		return ID;
	}

	public synchronized Boolean executeUpdate(String sql) {
		if (isStop())
			return null;
		/** 影响的行数 **/
		boolean ID = false;
		setConn();
		try {
			/** 从数据库连接池中获取数据库连接 **/
			/** 执行SQL预编译 **/
			Statement st;
			_st = st = _myconn.createStatement(); // (sql, Statement.RETURN_GENERATED_KEYS);
			ID = st.execute(sql);
			_commitcount.incrementAndGet();
		} catch (Exception e) {
			on执行异常("executeUpdate(sql); sql=" + sql, e);
		}
		autoCloseAll();
		return ID;
	}

	/**
	 * 批量执行:适用于sql不变,而 参数连续改变的情况 适用于大数据插入 executeBatch 之后无法执行其他语句 注意 本函数是
	 * 立即自动提交,设置手动提交,只是能够继续调用,无需重新实例化或重新连接
	 */
	public synchronized Integer batch(String sql, List<Object[]> sqlparams) {
		if (isStop())
			return -1;
		/** 影响的行数 **/
		Integer affectRowCount = 0;
		setConn();
		try {
			/** 设置不自动提交， 使用事物 ，以便于在出现异常的时候数据库回滚 **/
			/** 执行SQL预编译 **/
			PreparedStatement pst;
			_st = pst = _myconn.prepareStatement(sql);
			/** 从数据库连接池中获取数据库连接 **/
			for (Object[] bindArgs : sqlparams) {
				if (bindArgs != null) {
					/** 绑定参数设置sql占位符中的值 **/
					for (int i = 0; i < bindArgs.length; i++) {
						pst.setObject(i + 1, bindArgs[i]);
					}
					pst.addBatch();
				}
				affectRowCount++;
			} /** 执行sql **/
			_commitcount.addAndGet(affectRowCount);
			pst.executeBatch();
		} catch (Exception e) {
			on执行异常("batch(sql,List<[]>); sql=" + sql, e);
		}
		autoCloseAll();
		return affectRowCount;
	}

	/**
	 * 批量执行: 适用于 执行大量 无参数sql, 参数写入sql中 executeBatch 之后无法执行其他语句 注意 本函数是
	 * 立即自动提交,设置手动提交,只是能够继续调用,无需重新实例化或重新连接
	 */
	public synchronized Integer batch(List<String> sqls) {
		if (isStop())
			return -1;
		/** 影响的行数 **/
		Integer affectRowCount = 0;
		Statement st = null;
		setConn();
		try {
			/** 执行SQL预编译 **/
			_st = st = _myconn.createStatement();
			/** 从数据库连接池中获取数据库连接 **/
			for (String sql : sqls) {
				if (MyFun.isEmpty(sql))
					continue;
				st.addBatch(sql);
				affectRowCount++;
			} /** 批量执行sql **/
			_commitcount.addAndGet(affectRowCount);
			st.executeBatch();
		} catch (Exception e) {
			on执行异常("batch(list<sql>);", e);
		}
		autoCloseAll();
		return affectRowCount;
	}

	/**
	 * 定时提交周期
	 */
	private static final int PERIOD_MS = 2000;
	private MyTimer _mtimer = null;

	/** 每隔一段时间自动提交,定时周期(毫秒为单位) */
	public void timeoutCommit(Integer periodms) {
		if (periodms == null)
			periodms = PERIOD_MS;
		if (_mtimer == null) {
			_mtimer = MyTimer.getInstance();
			// System.out.println("mydb.timeoutCommit()="+_mtimer.getName());
			// 每隔两个小时重试
			_mtimer.start(new Runnable() {
				@Override
				public void run() {
					// 提交
					commit();
				}// periodms
			}, 1000, periodms);
		}
	}

	public static List<String> createDatabaseSql(String dbname) {
		List<String> sqls = new ArrayList<>();
		sqls.add("DROP DATABASE  " + dbname);
		sqls.add("CREATE DATABASE " + dbname + " CHARACTER SET utf8 ");
		return sqls;
	}
}
/*
 * 原理
 * 
 * 回滚: 回退截止目前的所有有效语句
 * 
 * 提交: 提交截止目前未提交的所有有效语句
 * 
 * 执行异常: 代码只会抛出异常,但是不做处理.不会影响其他正常sql 的commit()
 *
 * 容易错误的理解:
 * 
 * 异常必须回退 错误,只有在事物中才需要回退
 * 
 * 语句出现异常 与 commit和rollback 无任何关系
 * 
 * commit和rollback 出现异常通常与connect 或statment 有关
 * 
 * select 如果在 commit 之前执行 ,是不会受到commit 影响的
 * 
 * 一定需要注意的:
 * 
 * commit 方法中 如果 提交前有异常语句,但是 正常的语句还是会写入内存,
 * 
 * 这个时候(commit之前)select ,还是会查出数据的.,只有在commit之后才不会select出数据
 * 
 * 
 * 
 * 设计原则:
 * 
 * 1. db 所有手动或自动提交后必须调用关闭, 关闭以后会,连接返回连接池,如果发生异常,则抛弃不返回连接池 ;
 * 
 * 2. db auto , commit 和 timer 模式都可以 连续调用脚本 但是意义不同 auto 无事物效果, commit
 * 有事物效果,timer 有事物和自动提交效果
 * 
 * 3.提交前(包含手动或自动提交) 如果这时候连接是异常的,那么不做处理,直接抛弃连接,抛出异常
 * 
 * 4.处理sql前检查 连接,如果异常可以重新从连接池获取 处理后 手动返回连接池
 * 
 * 用法: commit 方式提交,如果错一个 所有都无法提交
 * 
 * auto 方式提交,除了错的,其他都可以正常提交
 * 
 * timer 方式提交,错之前未提交的 都无法提交,后面正常的可以提交
 * 
 * batch 方式 中间出错,整体无法提交,如果结合commit使用 ,所有未提交的 sql 都无法提交
 * 
 */